#pragma once
#include <iostream>
#include <string>
#include <list>
#include <memory>
#include <mutex>
#include "../mqcommom/helper.hpp"
#include "../mqcommom/msg.pb.h"
/************************
 * 将消息在文件中管理,知道偏移量就能插入，考虑粘包问题(消息的长度)
 *  消息管理是基于对各自的队列管理，指定队列的名称,一个队列，创建一个文件管理,多个文件统一放在目录下(basedir)
 * 对外的接口
 *      1.创建目录，创建指定文件。
 *      2.往指定的文件里，插入一条消息,插入先插入4字节消息的长度，再插入消息body(序列化后),最后更新
 *      3.消息的删除,只需要将覆盖写，将消息上的有效位转为无效
 *      4.垃圾回收机制:总消息超过2000，有一半消息无效
 *             1)获取所有的有效消息 2)往临时文件里写数据 3)更新消息的偏移量 4)删除源文件，对临时文件重命名
 *
 * MesgQueue:以队列管理消息
 *      1.mapper的操作句柄，管理文件 2.用list管理内存中数据 3.map映射持久化数据(为了快速查找，不用list) 4.待确认数据(没有收到ack) 5.队列名称
 * 对外接口:
 *      1)消息恢复 2)插入mesage 在内存中添加,如果有持久化，在文件中也添加。 3)获取队头的消息
 *      4)移除数据:移除待确认数据寻找，如果持久化，那么就要垃圾回收，移除删除的管理
 *      5)垃圾回收:总消息超过2000，有一半消息无效 ,获取所有的有效mesageptr,更新内存中的ptr映射
 *
 * MessageQueueManager的管理
 *      1.利用map管理每一个队列
 *      2.对外提供接口:初始化(创建队列):如果队列存在，就退出，队列不存在，就构建智能指针并插入
 *              新增/删除消息:加锁的时机，对queue_msgs加锁 如果不存在句柄就不操作
 *              获取队列 ack(删除消息) :加锁
 ********************/
using namespace ns_helper;
namespace ns_rabbitmq
{
    // data/queue1.mqd data/queue1.mqd.tmp
    static const std::string datafile_subfix = ".mqd";
    static const std::string tmpfile_subfix = ".mqd.tmp";
    using MessagePtr = std::shared_ptr<ns_rabbitmq::Message>;
    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            // 创建目录
            if (basedir.back() != '/')
                basedir += "/";
            FileHelper::CreateDirctory(basedir);
            _datafile = basedir + qname + datafile_subfix;
            _tmpfile = basedir + qname + tmpfile_subfix;
            if (!FileHelper(basedir).IsExists())
                assert(FileHelper::CreateDirctory(basedir));

            assert(CreateFilePath());
        }

        bool CreateFilePath()
        {
            if (FileHelper(_datafile).IsExists())
                return true;
            if (!FileHelper::CreateFile(_datafile))
                ELOG("create file %s error", _datafile.c_str());
            return true;
        }
        void RemoveFile()
        {
            FileHelper::RemoveFile(_datafile);
            // FileHelper::RemoveFile(_tmpfile);
            //
            DLOG("删除双文件%s,%s", _datafile.c_str(), _tmpfile.c_str());
        }

        bool Insert(MessagePtr &msgptr)
        {
            return Insert(_datafile, msgptr);
        }

        bool Remove(MessagePtr &msgptr)
        {
            // 覆盖写
            msgptr->mutable_payload()->set_valid("0");
            std::string body = msgptr->payload().SerializeAsString();
            // 获取偏移量和长度
            size_t fseek = msgptr->offset();
            size_t lenth = msgptr->length();
            if (body.size() != lenth)
            {
                ILOG("覆盖后长度不一致,移除失败");
                return false;
            }

            FileHelper file_helper(_datafile);
            bool ret = file_helper.Write(body.c_str(), fseek, lenth);
            if (!ret)
            {
                ILOG("write body to file error");
                return false;
            }
            return true;
        }

        std::list<MessagePtr> GarbageRecycl()
        {
            // 1.加载所有的有效message
            std::list<MessagePtr> result;
            int ret = LoadMessage(result);
            if (!ret)
            {
                DLOG("loading message fail");
                return std::list<MessagePtr>();
            }
            DLOG("loading message size %d", FileHelper(_datafile).Size());
            // 2.创建临时文件
            FileHelper::CreateFile(_tmpfile);
            FileHelper file_help(_tmpfile);
            // 3.将消息拷贝进临时文件
            for (auto &msg : result)
            {
                bool ret = Insert(_tmpfile, msg);
                if (!ret)
                {
                    DLOG("insert into tmp file error");
                    return result;
                }
            }

            // 删除源文件
            if (!FileHelper::RemoveFile(_datafile))
            {
                return result;
            }
            // 重命名
            if (!file_help.RenameFile(_datafile))
            {
                return result;
            }

            return result;
        }

    private:
        bool LoadMessage(std::list<MessagePtr> &result)
        {
            int fseek = 0, msg_size;
            FileHelper file_help(_datafile);
            msg_size = file_help.Size();
            while (fseek < msg_size)
            {
                // 1. 一直读取数据
                size_t lenth;
                bool ret = file_help.Read((char *)&lenth, fseek, sizeof(lenth));
                if (!ret)
                {
                    ELOG("read lenth from file:%s error", _datafile.c_str());
                    return false;
                }
                // 2. 获取body
                fseek += sizeof(lenth);
                std::string body(lenth, '\0');
                ret = file_help.Read(&body[0], fseek, lenth);
                if (!ret)
                {
                    ELOG("read body from file:%s error", _datafile.c_str());
                    return false;
                }
                fseek += lenth;
                // 3.构建消息
                MessagePtr msgptr = std::make_shared<Message>();
                msgptr->mutable_payload()->ParseFromString(body);
                // 4.获取标志
                if (msgptr->payload().valid() == "0")
                {
                    continue;
                }
                result.push_back(msgptr);
            }

            return true;
        }

        bool Insert(const std::string &filename, MessagePtr &msgptr)
        {
            // 1.获取序列化后的body,长度
            // 往尾部write长度 / body
            // 设置偏移量
            std::string body = msgptr->payload().SerializeAsString();
            // 获取长度
            size_t fsize = body.size();
            // 获取偏移量
            FileHelper file_helper(filename);
            size_t fseek = file_helper.Size();
            // 2.先写入4字节的长度，再写入body
            bool ret = file_helper.Write((char *)&fsize, fseek, sizeof(fsize));
            if (!ret)
            {
                return false;
            }
            fseek += sizeof(fsize);
            ret = file_helper.Write(body.c_str(), fseek, fsize);
            if (!ret)
            {
                return false;
            }
            // 3.更新偏移量和长度
            msgptr->set_offset(fseek);
            msgptr->set_length(fsize);
            return true;
        }

    private:
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    class MessageQueue
    {
    public:
        MessageQueue(std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname), _name(qname), _total_msg(0), _valid_count(0)
        {
        }
        void Recovry()
        {
            // 主要是获取mapper的数据,更新内存中的数据
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.GarbageRecycl();
            for (auto &msg : _msgs)
            {
                _durable_msg.insert(std::make_pair(msg->payload().basic_properties().id(), msg));
            }
            _valid_count = _total_msg = _msgs.size();
        }

        bool Insert(const BasicProperties *basicProp, const std::string &body, bool durable)
        {
            // 1.构建对象
            MessagePtr msgptr = std::make_shared<Message>();
            msgptr->mutable_payload()->set_body(body);
            DeliveryMode mod;
            if (durable)
                mod = DeliveryMode ::DURABLE;
            else
                mod = DeliveryMode ::UNDURABLE;
            if (basicProp)
            {
                // 用户输入的是正确的属性
                msgptr->mutable_payload()->mutable_basic_properties()->set_id(basicProp->id());
                msgptr->mutable_payload()->mutable_basic_properties()->set_routing_key(basicProp->routing_key());
            }
            else
            {
                msgptr->mutable_payload()->mutable_basic_properties()->set_id(UUIDHelp::uuid());
                msgptr->mutable_payload()->mutable_basic_properties()->set_routing_key("");
            }
            msgptr->mutable_payload()->mutable_basic_properties()->set_deliver_mode(mod);

            std::unique_lock<std::mutex> lock(_mutex);

            // 3.设置了持久化
            if (durable)
            {
                msgptr->mutable_payload()->set_valid("1");
                bool ret = _mapper.Insert(msgptr);
                if (!ret)
                {
                    DLOG("往文件添加进消息失败");
                    return false;
                }
                ILOG("持久化了%s", msgptr->payload().basic_properties().id().c_str());
                _valid_count++;
                _total_msg++;
                _durable_msg.insert(std::make_pair(msgptr->payload().basic_properties().id(), msgptr));
            }

            // 2.插入到内存中
            _msgs.push_back(msgptr);
            return true;
        }

        MessagePtr GetFront()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_msgs.size())
                return MessagePtr();
            // 1.将数据添加进待确认
            MessagePtr msgptr = _msgs.front();
            _msgs.pop_front();
            _wait_ack_msg.insert(std::make_pair(msgptr->payload().basic_properties().id(), msgptr));
            // 2.返回队头
            return msgptr;
        }

        bool Remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _wait_ack_msg.find(msg_id);
            if (it == _wait_ack_msg.end())
            {
                ILOG("确认消息队列没有指定%s消息", msg_id.c_str());
                return false;
            }
            // 删除持久化
            if (it->second->payload().basic_properties().deliver_mode() == DeliveryMode ::DURABLE)
            {
                ILOG("从文件中删除%s", msg_id.c_str());
                _mapper.Remove(it->second);
                _durable_msg.erase(msg_id);
                --_valid_count;
                GC();
            }

            _wait_ack_msg.erase(msg_id);

            return true;
        }

        size_t SandableSize()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t TotalMsgSize()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_msg;
        }
        size_t DurableSize()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msg.size();
        }
        size_t WaitAckSize()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _wait_ack_msg.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.RemoveFile();
            _durable_msg.clear();
            _msgs.clear();
            _wait_ack_msg.clear();
            _total_msg = _valid_count = 0;
        }

    private:
        void GC()
        {
            bool ret = CheackGC();
            if (!ret)
                return;
            std::list<MessagePtr> msgs = _mapper.GarbageRecycl();
            // 更新内存中的映射
            for (auto &msg : msgs)
            {
                auto it = _durable_msg.find(msg->payload().basic_properties().id());
                if (it == _durable_msg.end())
                {
                    ILOG("有一条持久化数据%s,没有被记录在内存中", msg->payload().basic_properties().id().c_str());
                    _durable_msg.insert(std::make_pair(msg->payload().basic_properties().id(), msg));
                    _msgs.push_back(msg);
                    continue;
                }
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 更新数据的总数
            _valid_count = _total_msg = msgs.size();
        }

        bool CheackGC()
        {
            if (_total_msg > 2000 && _valid_count * 10 / _total_msg < 5)

            {
                std::cout << "垃圾回收" << std::endl;
                return true;
            }
            return false;
        }

    private:
        std::mutex _mutex;
        MessageMapper _mapper;
        std::string _name;
        size_t _total_msg;
        size_t _valid_count;
        std::list<MessagePtr> _msgs;
        std::unordered_map<std::string, MessagePtr> _durable_msg;
        std::unordered_map<std::string, MessagePtr> _wait_ack_msg;
    };

    using QueueMsgPtr = std::shared_ptr<MessageQueue>;
    class MessageManager
    {
    public:
        using MessageManagerPtr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string &basedir)
            : _basedir(basedir) {}

        void InitQueueMsg(const std::string &qname)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it != _msg_queues.end())
                {
                    return;
                }
                queptr = std::make_shared<MessageQueue>(_basedir, qname);
                _msg_queues.insert(std::make_pair(qname, queptr));
            }
            queptr->Recovry();
        }

        void DestoryQueueMsg(const std::string &qname)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                    DLOG("队列%s不存在", qname.c_str());
                    return;
                }
                queptr = it->second;
                _msg_queues.erase(it);

                DLOG("清理完内存中的队列%s", qname.c_str());
            }
            queptr->clear();
        }
        MessagePtr Front(const std::string &qname)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                   DLOG("队列%s不存在", qname.c_str());
                    return MessagePtr();
                }
                queptr = it->second;
            }
            return queptr->GetFront();
        }

        bool Insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                    DLOG("队列%s不存在", qname.c_str());
                    return false;
                }
                queptr = it->second;
            }
            return queptr->Insert(bp, body, queue_is_durable);
        }

        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                    DLOG("队列%s不存在", qname.c_str());
                    return;
                }
                queptr = it->second;
            }
            queptr->Remove(msg_id);
        }

        size_t GetSendable_count(const std::string &qname)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                    DLOG("队列%s不存在", qname.c_str());
                    return 0;
                }
                queptr = it->second;
            }
            return queptr->SandableSize();
        }

        size_t GetTotal_count(const std::string &qname)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                    DLOG("队列%s不存在", qname.c_str());
                    return 0;
                }
                queptr = it->second;
            }
            return queptr->TotalMsgSize();
        }

        size_t GetDurable_count(const std::string &qname)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                    DLOG("队列%s不存在", qname.c_str());
                    return 0;
                }
                queptr = it->second;
            }
            return queptr->DurableSize();
        }

        size_t GetWaitAck_count(const std::string &qname)
        {
            QueueMsgPtr queptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _msg_queues.find(qname);
                if (it == _msg_queues.end())
                {
                    DLOG("队列%s不存在", qname.c_str());
                    return 0;
                }
                queptr = it->second;
            }
            return queptr->WaitAckSize();
        }

        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &msg : _msg_queues)
            {
                msg.second->clear();
            }
            _msg_queues.clear();
        }

    private:
        std::string _basedir;
        std::mutex _mutex;
        std::unordered_map<std::string, QueueMsgPtr> _msg_queues;
    };
}